package blue.starter.internal.mqtt.config;

import blue.starter.internal.core.util.BeanFactoryUtil;
import blue.starter.internal.mqtt.consumer.DefaultMqttConsumer;
import blue.starter.internal.mqtt.consumer.MqttListenerConfig;
import blue.starter.internal.mqtt.producer.DefaultMqttProducer;
import blue.starter.internal.mqtt.producer.MqttClient;
import blue.starter.internal.mqtt.property.MqttConfigProperties;
import blue.starter.internal.mqtt.property.MqttProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.RootBeanDefinition;

/**
 * @author Jin Zheng
 * @since 1.0 2020-12-31
 */
public class MqttBeanPostProcessor implements BeanFactoryPostProcessor, BeanPostProcessor, BeanFactoryAware
{
	private static Logger logger = LoggerFactory.getLogger(MqttBeanPostProcessor.class);

	private MqttProperties mqttProperties;
	private BeanFactory beanFactory;
	private BeanDefinitionRegistry registry;

	public MqttBeanPostProcessor()
	{
	}

	@Override
	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException
	{
		if (mqttProperties != null)
		{
			synchronized (this)
			{
				if (mqttProperties != null)
				{
					return bean;
				}
			}
		}

		mqttProperties = beanFactory.getBean(MqttProperties.class);
		if (mqttProperties == null || mqttProperties.getConfigs() == null)
			return bean;

		for (MqttConfigProperties config : mqttProperties.getConfigs())
		{
			this.registryMqttClient(config);
			this.registryMqttProducer(config);
			this.registryMqttConsumer(config);
		}
		return bean;
	}

	private void registryMqttClient(MqttConfigProperties config)
	{
		RootBeanDefinition definition = new RootBeanDefinition(MqttClient.class);
		MutablePropertyValues propertyValues = definition.getPropertyValues();
		propertyValues.addPropertyValue("name", config.getName());
		propertyValues.addPropertyValue("broker", config.getBroker());
		propertyValues.addPropertyValue("username", config.getUsername());
		propertyValues.addPropertyValue("password", config.getPassword());
		propertyValues.addPropertyValue("clientId", config.getClientId());
		propertyValues.addPropertyValue("count", config.getCount());
		propertyValues.addPropertyValue("keepAlive", config.getKeepAlive());
		propertyValues.addPropertyValue("reconnectDelay", config.getReconnectDelay());
		registry.registerBeanDefinition(config.getName(), definition);
	}

	private void registryMqttProducer(MqttConfigProperties config)
	{
		RootBeanDefinition definition = new RootBeanDefinition(DefaultMqttProducer.class);
		MutablePropertyValues propertyValues = definition.getPropertyValues();
		propertyValues.addPropertyValue("name", config.getProducerName());
		propertyValues.addPropertyValue("defaultQos", config.getDefaultQos());
		BeanFactoryUtil.addPropertyRef(propertyValues, "mqttClient", config.getName());
		BeanFactoryUtil.addPropertyRef(propertyValues, "producerListener", config.getProducerListener());
		registry.registerBeanDefinition(config.getProducerName(), definition);
	}

	private void registryMqttConsumer(MqttConfigProperties config)
	{
		if (config.getListeners() == null || config.getListeners().isEmpty())
			return;

		RootBeanDefinition definition = new RootBeanDefinition(DefaultMqttConsumer.class);
		MutablePropertyValues propertyValues = definition.getPropertyValues();
		this.registryMqttListenerConfig(config, definition);
		propertyValues.addPropertyValue("name", config.getConsumerName());
		propertyValues.addPropertyValue("defaultQos", config.getDefaultQos());
		BeanFactoryUtil.addPropertyRef(propertyValues, "taskExecutor", config.getDefaultTaskExecutor());
		BeanFactoryUtil.addPropertyRef(propertyValues, "exceptionHandler", config.getDefaultExceptionHandler());
		BeanFactoryUtil.addPropertyRef(propertyValues, "mqttClient", config.getName());
		registry.registerBeanDefinition(config.getConsumerName(), definition);
	}

	private void registryMqttListenerConfig(MqttConfigProperties config, RootBeanDefinition definition)
	{
		ManagedList<BeanDefinitionHolder> managedList = new ManagedList<>();
		for (var listenerConfig : config.getListeners())
		{
			RootBeanDefinition listenerDefinition = new RootBeanDefinition(MqttListenerConfig.class);
			MutablePropertyValues propertyValues = listenerDefinition.getPropertyValues();
			propertyValues.addPropertyValue("topic", listenerConfig.getTopic());
			propertyValues.addPropertyValue("qos", listenerConfig.getQos());
			propertyValues.addPropertyValue("listener", new RuntimeBeanReference(listenerConfig.getBeanName()));
			propertyValues.addPropertyValue("multiThread", listenerConfig.isMultiThread());
			BeanFactoryUtil.addPropertyRef(propertyValues, "taskExecutor", listenerConfig.getTaskExecutor());
			BeanFactoryUtil.addPropertyRef(propertyValues, "exceptionHandler", listenerConfig.getExceptionHandler());
			BeanDefinitionHolder holder = new BeanDefinitionHolder(listenerDefinition, DefaultMqttConsumer.class.getName());
			managedList.add(holder);
		}
		definition.getPropertyValues().addPropertyValue("configList", managedList);
	}

	@Override
	public void setBeanFactory(BeanFactory beanFactory) throws BeansException
	{
		this.beanFactory = beanFactory;
	}

	@Override
	public void postProcessBeanFactory(ConfigurableListableBeanFactory factory) throws BeansException
	{
		this.registry = (BeanDefinitionRegistry) factory;
	}
}
